package com.lxn.app

import com.alibaba.fastjson.{JSON, JSONObject}
import com.lxn.until.{MYKafkaUtils, MyOffsetUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast

import java.util
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object OdsBaseDbApp {

  def main(args: Array[String]): Unit = {
    //创建业务数据的实时环境
    val sparkConf: SparkConf = new SparkConf().setAppName("ods_base_db_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf,Seconds(5))

    //创建主题和组
    val topicName : String = "ODS_BASE_DB"
    val groupId : String = "ODS_BASE_GROUP_DB"
    //读取偏移量
    val offsets: Map[TopicPartition, Long] = MyOffsetUtils.readOffset(groupId, topicName)
    //判断是否存在偏移量
    var offsetDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(offsets!=null && offsets.nonEmpty){
      //使用重载的指定偏移量的消费
      offsetDStream= MYKafkaUtils.getConsumer(ssc, topicName, groupId, offsets)
    }else{
      offsetDStream = MYKafkaUtils.getConsumer(ssc,topicName,groupId)
    }

    //转换数据格式之前我们需要将偏移量进行保存
    var offsetsRanges :Array[OffsetRange] = null
    val offsetRangeDStream : DStream[ConsumerRecord[String,String]] = offsetDStream.transform(
      rdd => {
        offsetsRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    //转换数据格式，因为这种数据格式是不能被传递的，它没有被序列化
    val jsonObjDStream = offsetRangeDStream.map(
      consumerRecord => {
        //这里我们只考虑数据的值
        val dataJson = consumerRecord.value()
        val jsonObj = JSON.parseObject(dataJson)
        jsonObj
      }
    )
    jsonObjDStream.print(100)

    //开始进行数据分流操作
    //这样写不方便代码维护，我们改一下
    /*val factTables : Array[String] = Array[String]("order_info","order_detail")
    val dimTables : Array[String] = Array[String]("user_info","base_province")*/
    jsonObjDStream.foreachRDD(
      rdd => {  //对流中每一个rdd进行处理
        //将表名放到redis存储
        val factKey : String = "FACT:TABLES"
        val dimKey : String = "DIM:TABLES"
        //获取连接对象sme

        val jedis1 = MyRedisUtils.getRedisFromPool()
        //从redis中获取数据 这个类型能够传递的话，那就是序列化的
        val factTables : util.Set[String] = jedis1.smembers(factKey)
        println("factTables: " + factTables)
        //如果数据太大，就不要再driver端创建，但是小的话没关系，不是太大我们可以使用广播变量
        val factTablesBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(factTables)

        val dimTables : util.Set[String] = jedis1.smembers(dimKey)
        println("dimTables: " + dimTables)
        val dimTablesBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(dimTables)
        jedis1.close()

        rdd.foreachPartition( //rdd对每个分区进行处理，由于后面需要刷写数据到磁盘，所以使用这个比较合适，因为我们有4个分区
          jsonObjItr => { //每个分区有不等的json对向，所以我们需要遍历对每一个对象进行处理
            //算子里面的代码都是在executor端执行的，所以我们需要在for循环外面，executor里面创建连接对象
            val jedis = MyRedisUtils.getRedisFromPool()
            for (jsonObj <- jsonObjItr) {
              //对每一个对象处理
              //观察数据的格式，我们选择用表名划分主题，同时主题加上操作类型
              val opeType: String = jsonObj.getString("type")
              val ope = opeType match {
                case "bootstrap-insert" => "I"
                case "insert" => "I"
                case "update" => "U"
                case "delete" => "D"
                case _ => null
              }
              if (ope != null) {
                //提取表名
                val tableName: String = jsonObj.getString("table")

                //判断表的类型 事实表
                if (factTablesBC.value.contains(tableName)) {
                  //提取数据
                  val data: String = jsonObj.getString("data")
                  val dwdTopicName: String = s"DWD_${tableName.toUpperCase()}_${ope}"
                  MYKafkaUtils.send(dwdTopicName, data)

                }
                //维度数据，需要存储至redis 考虑内存是否够
                if(dimTablesBC.value.contains(tableName)){
                  //提取数据
                  val dataJson : JSONObject = jsonObj.getJSONObject("data")
                  val id = dataJson.getString("id")
                  //拼接key
                  val redisKey : String = s"DIM:${tableName.toUpperCase()}:${id}"
                  //获取redis连接对象
                  //连接对象在for循环里面，一直不断关闭和连接，导致性能降低，所以我们找个合适位置
//                  val jedis = MyRedisUtils.getRedisFromPool()
                  jedis.set(redisKey,dataJson.toJSONString)
//                  jedis.close()
                }

              }

            }
            //关闭redis连接
            jedis.close()
            //刷新Kafka缓冲区
            MYKafkaUtils.flush()
          }
        )
        //保存偏移量
//        println("RDD处理完成，保存偏移量: " + offsetsRanges.map(o => s"${o.partition}:${o.untilOffset}").mkString(","))
        MyOffsetUtils.saveOffset(groupId,topicName,offsetsRanges)
      }
    )

    ssc.start()
    ssc.awaitTermination()


  }
}
